package com.clw.dwd;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.clw.utils.MyKafkaUtil;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

/**
 * 3.2.3利用侧输出流实现数据拆分
 */
public class SideStreamApp {
    public static void main(String[] args) {
        // 1. 创建流环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // 获取kafka 的消费者的流
        FlinkKafkaConsumer<String> logSource = MyKafkaUtil.getKafkaConsumer("ods_base_log", "safsfsd");


        // 获取流数据
        DataStreamSource<String> logStream = env.addSource(logSource);

        // 过滤 数据清洗
        SingleOutputStreamOperator<JSONObject> commonLog = logStream.filter(x -> {  // 检测是否为合法的json字符串
            try {
                return true;
            } catch (Exception e) {
                return false;
            }
        }).map(x -> {
            return JSON.parseObject(x);
        });

        // 创建tag
        OutputTag<JSONObject> startTag = new OutputTag<JSONObject>("start") {
        };
        OutputTag<JSONObject> pageTag = new OutputTag<JSONObject>("page") {
        };
        OutputTag<JSONObject> displayTag = new OutputTag<JSONObject>("display") {
        };

        // 开始侧分
        SingleOutputStreamOperator<JSONObject> processStream = commonLog.process(new ProcessFunction<JSONObject, JSONObject>() {
            @Override
            public void processElement(JSONObject jsonObject, ProcessFunction<JSONObject, JSONObject>.Context context, Collector<JSONObject> collector) throws Exception {
                if (jsonObject.containsKey("page")) {
                    // 输出到侧流
                    context.output(pageTag, jsonObject);
                }
                if (jsonObject.containsKey("start")) {
                    // 输出到侧流
                    context.output(startTag, jsonObject);

                }
                if (jsonObject.containsKey("displays")) {
                    // 输出到侧流
                    context.output(displayTag, jsonObject);
                }

                if (jsonObject.containsKey("common")) {
                    // 输出到主流
                    collector.collect(jsonObject);
                }
            }
        });

        DataStream<JSONObject> streamDisaply = processStream.getSideOutput(displayTag);
        DataStream<JSONObject> streamStart = processStream.getSideOutput(startTag);
        DataStream<JSONObject> streampage = processStream.getSideOutput(pageTag);
//        streamDisaply.print();
//        streamStart.print();
//        streampage.print();
        // 输出主流
        processStream.print();

    }

}
